package com.sf.iisp.kafka.adapter;

import com.sf.kafka.api.produce.ProduceConfig;
import com.sf.kafka.api.produce.ProduceOptionalConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

public class KafkaProducerFactoryBean implements FactoryBean<IKafkaProducerPoolDelegate>, InitializingBean {

    private IKafkaProducerPoolDelegate producerPoolDelegate;

    private int poolSize = 1;
    private String monitorUrl;
    private String clusterName;
    private String topic;
    private String checkCode;
    private ProduceOptionalConfig optionalConfig;

    @Override
    public IKafkaProducerPoolDelegate getObject() throws Exception {
        return producerPoolDelegate;
    }

    @Override
    public Class<?> getObjectType() {
        return IKafkaProducerPoolDelegate.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.notNull(monitorUrl, "monitorUrl can not be null!");
        Assert.notNull(clusterName, "clusterName can not be null!");
        Assert.notNull(topic, "topic can not be null!");
        Assert.notNull(checkCode, "checkCode can not be null!");

        final String topicTokens = topic + ":" + checkCode;
        ProduceConfig produceConfig = new ProduceConfig(poolSize, monitorUrl, clusterName, topicTokens);
        //kafkaProducer = new ProducerPoolDelegate(produceConfig, optionalConfig);
        producerPoolDelegate = new KafkaProducerPoolDelegate(produceConfig, optionalConfig);
    }

    public void setPoolSize(int poolSize) {
        this.poolSize = poolSize;
    }

    public void setMonitorUrl(String monitorUrl) {
        this.monitorUrl = monitorUrl;
    }

    public void setClusterName(String clusterName) {
        this.clusterName = clusterName;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public void setCheckCode(String checkCode) {
        this.checkCode = checkCode;
    }

    public void setOptionalConfig(ProduceOptionalConfig optionalConfig) {
        this.optionalConfig = optionalConfig;
    }
}
